package com.codejiwei.flink.practice;

import com.codejiwei.flink.entity.AdsClickLog;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * @author jiwei
 * @description
 * @date 2023/5/23 9:53
 */
public class Flink_Project_AdsClick {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);

        env.readTextFile("E:\\workspace\\ss-flink\\src\\main\\resources\\data\\AdClickLog.csv")
                .map(line -> {
                    String[] fields = line.split(",");
                    return new AdsClickLog(
                            Long.valueOf(fields[0]),
                            Long.valueOf(fields[1]),
                            fields[2],
                            fields[3],
                            Long.valueOf(fields[4])
                    );
                })
                .map(ads -> Tuple2.of(Tuple2.of(ads.getAdId(), ads.getProvince()), 1L))
                .returns(Types.TUPLE(Types.TUPLE(Types.LONG, Types.STRING), Types.LONG))
                .keyBy(new KeySelector<Tuple2<Tuple2<Long, String>, Long>, Tuple2<Long, String>>() {
                    @Override
                    public Tuple2<Long, String> getKey(Tuple2<Tuple2<Long, String>, Long> tuple2LongTuple2) throws Exception {
                        return tuple2LongTuple2.f0;
                    }
                })
                .sum(1)
                .print("省份-广告");

//                .map(ads -> Tuple3.of(ads.getAdId(), ads.getProvince(), 1L))
//                .returns(Types.TUPLE(Types.LONG, Types.STRING, Types.LONG))
//                .keyBy(new KeySelector<Tuple3<Long, String, Long>, Tuple2<Long, String>>() {
//                    @Override
//                    public Tuple2<Long, String> getKey(Tuple3<Long, String, Long> longStringLongTuple3) throws Exception {
//                        return Tuple2.of(longStringLongTuple3.f0, longStringLongTuple3.f1);
//                    }
//                })
//                .sum(2)
//                .print();

        env.execute();
    }
}
